Reapply "Add ExecutionPlan::apply_expressions() (apache#20337)" (apache#22437) - #24018
Conversation
github-actions
Bot
added
documentation
jayshrivastava
changed the title
|
Thank you for opening this pull request! Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch). Details |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #24018 +/- ##
==========================================
- Coverage 81.06% 80.98% -0.08%
==========================================
Files 1106 1106
Lines 382025 382915 +890
Branches 382025 382915 +890
==========================================
+ Hits 309671 310120 +449
- Misses 54076 54493 +417
- Partials 18278 18302 +24 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
b8794d2 to
5514194
Compare
| /// `apply_expressions` + `downcast_ref::<DynamicFilterPhysicalExpr>` and | ||
| /// counting nodes. Neither API is observable from SQL. | ||
| #[tokio::test] | ||
| async fn test_discover_dynamic_filters_via_expressions_api() { |
There was a problem hiding this comment.
This test explicitly covers that dynamic filters are discoverable post-pushdown. This is what we want in datafusion-distributed and I assume other projects want this as well
| aggregate.with_dynamic_filter_expr(dynamic_filter)? | ||
| } else { | ||
| let mut aggregate = aggregate; | ||
| aggregate.dynamic_filter = None; |
There was a problem hiding this comment.
Semi-related. I think this was a bug. We kept the default dynamic filter created by try_new around even there's no pushed down filter. Now, we explicitly remove it. The test I added in datafusion/proto/tests/cases/roundtrip_physical_plan.rs covers this.
|
I see the code coverage is isn't very high. I don't see any good candidates to call |
gene-bordegaray
left a comment
gene-bordegaray
left a comment
There was a problem hiding this comment.
I had some questions about the exact contract but overall I think this almost there, just some clarification and looking very clean 👍
| let mut found = false; | ||
| plan.apply(|node| { | ||
| node.apply_expressions(&mut |root| { | ||
| root.apply(|expr| { |
There was a problem hiding this comment.
is there any way to eliminate this triple nesting we do.
From my understanding its:
- walking the execution plan tree
- Walk the expressions roots in one plan node
- Walks inside the one expression
Maybe something like a apply_plan_expressions would make this easier to read?
There was a problem hiding this comment.
I think this pattern is expected. I followed these docs for apply_expressions on logical plans to implement this for physical plans:
datafusion/datafusion/expr/src/logical_plan/plan.rs
Lines 135 to 144 in 5514194
| &self, | ||
| _f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>, | ||
| ) -> Result<TreeNodeRecursion> { | ||
| Ok(TreeNodeRecursion::Continue) |
There was a problem hiding this comment.
Since we already have apply_expression_roots to abstract "walk these expressions" case, would it be worth a small sibling helper for the "I have none" case? Right now that's just Ok(TreeNodeRecursion::Continue) repeated at every leaf/no-op node. Wondering if it would make sense to have something like apply_expression_continue() living next to it, just a plain function.
I think something like apply_expression_roots(std::iter::empty(), f) does the same thing, but since we're trying to abstract away TreeNodeRecursion, wondering if it would still make sense to have the dedicated helper
There was a problem hiding this comment.
apply_expression_roots(std::iter::empty(), f) is nice, but you have to use type annotations in practice:
apply_expression_roots(
std::iter::empty::<&Arc<dyn PhysicalExpr>>(),
f,
)
This is a bit less clean so
I added a helper apply_no_expressions instead.
There was a problem hiding this comment.
Jay (or anyone), feel free to disagree. I suggested this because imo it’s clearer to me, but whatever is clearest when upgrading is best.
| // So here, we try to use ref count to determine if the dynamic filter | ||
| // has actually be pushed down. | ||
| // Issue: <https://github.com/apache/datafusion/issues/18856> | ||
| let child_accepts_dyn_filter = Arc::strong_count(dyn_filter) > 1; |
There was a problem hiding this comment.
Even if the rewrite to use apply_expressions looks like it adds a bunch of code, I think this is a nice use case for it, relying on the strong count of the dynamic filter's inner/outer Arc has caused several headaches before. Maybe we could do something similar in HashJoinExec to determine whether the dynamic filter has a consumer? Right now that's done via is_used, which relies on strong count. IIRC it's only used to check for consumers, so maybe we could remove is_used entirely in favor of this approach?
There was a problem hiding this comment.
This is a good idea! I'll do it rn
|
This PR as it stands is in conflict with #23494. I think it may need a pass over it to align with the |
7550339 to
db63637
Compare
|
@cetra3 Imo, apply_expressions isn't leaking internal state, at least not to the extent prior to #23494. It's just providing iteration over expressions in a plan. It also mirrors the API for logical plan nodes datafusion/datafusion/expr/src/logical_plan/tree_node.rs Lines 418 to 421 in db63637 |
gene-bordegaray
left a comment
gene-bordegaray
left a comment
There was a problem hiding this comment.
last thing, other than that I think this is good 👍
There was a problem hiding this comment.
I believe should we apply expressions to this op because of this as well?
There was a problem hiding this comment.
it is done at line 315 it seems 🤔
There was a problem hiding this comment.
yes but is using apply_no_expressions(f)
I was thinking that since this method can add the ordering to the cached self.cache.eq_properties we would traverse that in the apply_expressions() method here.
Maybe I am still unclear about what needs to be traversed and not @jayshrivastava ?
There was a problem hiding this comment.
Good point. The behavior in this PR is inconsistent, so I just pushed a commit with a comment to define the behavior: I think we should ignore expressions in the properties() for several reasons
- It's a lot of code to have every
ExecutionPlanhas to visit the sort expr and the partitioning expr in the properties - Many
ExecutionPlannodes just returnself.children().properties(), so it's redundant - I doubt users will want
apply_expressionsto yield expressions in properties. I think users would only care about expressions that are evaluated and important during execution. - You can already call
properties()if you want the properties
Here's the updated doc comment:
/// Apply a closure `f` to each root expression that this node owns and uses
/// during execution, either by evaluating it or updating it dynamically.
///
/// An expression must not be visited solely because it describes an input or
/// output property, such as cached ordering, partitioning, or equivalence
/// metadata. However, these may be traversed indirectly. For example,
/// `RepartitionExec` visits the partitioning expressions it evaluates and
/// `SortExec` visits the sort expressions it evaluates to order rows.
///
/// This method is shallow: it must not visit expression children or expressions
/// owned by child execution plans.
///
/// Similarly to other [`TreeNode`] APIs, the closure can return
Let me know if this makes sense.
There was a problem hiding this comment.
yes this cleras things up, was moslty getting confused from documentationon when exactly we plan to use this and not but this clears things up 👍
There was a problem hiding this comment.
Good point. The behavior in this PR is inconsistent, so I just pushed a commit with a comment to define the behavior: I think we should ignore expressions in the properties() for several reasons
I agree -- I think it is unlikely that expressions will appear in properties that do not also appear in other expressions of the node (e.g. the exprs in a ProjectionExec)
There was a problem hiding this comment.
Also, the Exprs in properties are not actually evaluated because they are part of the properites, but rather are book keeping the result of that evaluation.
alamb
left a comment
alamb
left a comment
There was a problem hiding this comment.
Thank you @jayshrivastava and @gene-bordegaray -- I think this is looking quite nice and ready to go
Since I think we are under some time pressure, I took the liberty of merging up from main and resolving conflicts, and tweaking some documentation
I will also make a proposal PR to reduce the
datafusion_physical_plan::apply_expression_roots(
self.required_input_ordering
.iter()
.flatten()
.map(|sort_expr| &sort_expr.expr),
f,
)Boiler plate code, but we can also do that as a follow on
We might also be able to make this pattern easier to read
datafusion_physical_plan::apply_expression_roots(
self.projection
.source
.iter()
.map(|proj_expr| &proj_expr.expr),
f,
)By changing apply_expression_roots to take AsRef<Arc<PhysicalExpr>> and then implementing AsRef for ProjectionExpr
I think it could be collapsed into
datafusion_physical_plan::apply_expression_roots(
self.projection
.source
.iter()
f,
)| slice of the last non-empty input batch) and pass it to window expressions | ||
| via `WindowEvalContext::with_most_recent_row` instead of copying it into | ||
| each partition's state. | ||
| Add `apply_expressions` to your implementation. Call `f` on each top-level |
There was a problem hiding this comment.
This is fine, though it is technically redundant with the (very nice docs) on apply_expressions
There was a problem hiding this comment.
Specifically the examples are redundant I think
|
Sorry hit submit too early -- I am finishing my review now |
|
@alamb Thanks. I'll let you finish the review. Also I can fix up the commits! I figured I'd wait for a review first |
alamb
left a comment
alamb
left a comment
There was a problem hiding this comment.
TLDR looks good to me. I had some small simplification suggestions but I don't think they are required
As I mentioned, I did merge up from main and tweak some comments, but otherwise. think this is ready to go.
Thanks again (and thanks to @LiaCastaneda for starting this work!)
| /// | ||
| /// See [`ExecutionPlan::apply_expressions`] for more details and implementation examples. | ||
| /// | ||
| /// [`ExecutionPlan::apply_expressions`]: datafusion_physical_plan::ExecutionPlan::apply_expressions |
There was a problem hiding this comment.
This is good to direct readers to ExecutionPlan::apply_expressions and then document that heavily
| snapshot: snapshot_fn_wrapper, | ||
| snapshot_generation: snapshot_generation_fn_wrapper, | ||
| is_volatile_node: is_volatile_node_fn_wrapper, | ||
| expression_id: expression_id_fn_wrapper, |
There was a problem hiding this comment.
this appears to be an unrelated change, but a good one
There was a problem hiding this comment.
it is done at line 315 it seems 🤔
| slice of the last non-empty input batch) and pass it to window expressions | ||
| via `WindowEvalContext::with_most_recent_row` instead of copying it into | ||
| each partition's state. | ||
| Add `apply_expressions` to your implementation. Call `f` on each top-level |
There was a problem hiding this comment.
Specifically the examples are redundant I think
Thanks @jayshrivastava -- I am done, though it appears I (claude) botched the merge. Will fix |
| partition_count, | ||
| generator_count | ||
| ); | ||
| Arc::make_mut(&mut self.cache).partitioning = partitioning; |
There was a problem hiding this comment.
would this also be in the same boat? we would need to traverse these?
|
Addressed comments. I have to run but here are the remaining open threads of work/discussion
|
|
I am merging this one in so we can have downstream projects test with it briefly Thanks again @jayshrivastava and @gene-bordegaray and @LiaCastaneda |
A small quick follow on PR would be good too |
|
Should |
@alamb ok sounds good, thank you and we will be testing this as well 👍 |
|
Thanks for merging! |
7 participants
Which issue does this PR close?
ExecutionPlannodes discoverable #23814This change does not close the above issues because it does not implement a way to tell if a node is a producer dynamic filters.
Rationale for this change
See #23814 and datafusion-contrib/datafusion-distributed#553.
To send dynamic filter updates across the network, there needs to be a way to get access to
PhysicalExprfromExecutionPlan. As discussed in #23814, the cleanest way to do this is to addExecutionPlan::apply_expressions, which mirrors a similar method for logical plan nodes.What changes are included in this PR?
There's 3 commits in this PR:
Firstly, commit 1 re-applies the changes in #20337 (reverted in #22437).
Some of the reasons for why the original PR was reverted include
(a)
apply_expressionsis too complicated to implement and there's no concrete need to justify this complexity(b) there was no usage of
apply_expressionsinside this repoTo address (a)
ExecutionPlannodes discoverable #23814apply_expression_rootsandapply_no_expressionswhich abstract away theTreeNodeRecursioncomplexity from implementors. Now,apply_expressionsvery trivial to implement ex.&Arc<dyn PhysicalExpr>rather than&dyn PhysicalExprto reduce complexity around lifetimesTo address (b):
apply_expressionsinphysical-plan/src/aggregates/mod.rs. Previously, there was a hack that checked if a filter was pushed down usingArc::strong_count(dyn_filter) > 1. Now it usesapply_expressionsis_usedfrom dynamic filters which used to check Arc references counts to see if a filter was pushed down. Now, the hash join usesapply_expressionsto find pushed down filters.Are these changes tested?
Yes.
Are there any user-facing changes?
There's a new mandatory method
ExecutionPlan::apply_expressions(). See the upgrading guide and documentation for details.